package elasticsearchdeploylogic

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"strconv"
	"strings"
	"sync"
	"time"
	"yunzhan/common/utils"

	"yunzhan/common/models"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"
)

type DeployElasticsearchLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewDeployElasticsearchLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployElasticsearchLogic {
	return &DeployElasticsearchLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// DeployElasticsearch 部署Elasticsearch
func (l *DeployElasticsearchLogic) DeployElasticsearch(in *agent.DeployElasticsearchRequest) (*agent.DeployResponse, error) {
	if in.NodeInfo == nil {
		return &agent.DeployResponse{
			Code:    500,
			Message: "未配置节点信息, 无法执行Elasticsearch部署",
		}, errors.New("未配置节点信息, 无法执行Elasticsearch部署")
	}

	resp := &agent.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("请求已接收，正在部署 %s 到节点 %s", in.ConfigInfo.ComponentName, in.NodeInfo.Host),
	}

	go l.startDeployment(in)

	return resp, nil
}

func (l *DeployElasticsearchLogic) startDeployment(in *agent.DeployElasticsearchRequest) {

	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()

		// 创建 LogManager 实例
		logManager := models.NewLogManager(100, fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port))
		defer logManager.Close()

		flushInterval := 5 * time.Second // 刷新间隔
		maxBatchSize := 100              // 最大批量大小
		retryDelay := 2 * time.Second    // 重试延迟
		maxRetries := 3                  // 最大重试次数

		// 创建 ComponentInfoManager 实例
		componentInfoManager := models.NewComponentInfoManager(
			1000,
			fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port),
			flushInterval,
			maxBatchSize,
			retryDelay,
			maxRetries)
		defer componentInfoManager.Close()

		packageDirectory := in.ConfigInfo.PackagePath
		rtrimDir := utils.Rtrim(packageDirectory, "/")
		elasticsearchVersion := in.ConfigInfo.ComponentName
		installPath := utils.Rtrim(in.ConfigInfo.InstallPath, "/")
		baseInfo := fmt.Sprintf("安装包存放目录: %s, Elasticsearch version: %s, Elasticsearch安装目录: %s", rtrimDir, elasticsearchVersion, installPath)
		step := 1
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", baseInfo, "Elasticsearch", "Server")

		checkPathCommand := fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
`, installPath, installPath)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "执行组件安装路径检查...", "Elasticsearch", "Server")
		output, err := utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件安装路径异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件安装路径完成", "Elasticsearch", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包...", "Elasticsearch", "Server")
		tarCommand := fmt.Sprintf("sudo tar -xzf %s/%s -C %s", rtrimDir, elasticsearchVersion, installPath)
		output, err = utils.ExecCommand(tarCommand)
		if err != nil {
			message := fmt.Sprintf("解压安装包异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包完成", "Elasticsearch", "Server")

		ElasticsearchHome := utils.Rtrim(elasticsearchVersion, ".tar.gz")
		home := fmt.Sprintf("%s/%s", installPath, ElasticsearchHome)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查当前用户是否为root用户...", "Elasticsearch", "Server")
		if in.NodeInfo.Username == "root" {
			// 如果当前用户不是root，检查elasticsearch用户组是否存在
			checkGroupCommand := "getent group elasticsearch"
			output, err = utils.ExecCommand(checkGroupCommand)
			if err != nil || strings.TrimSpace(output) == "" {
				// 如果用户组不存在，创建elasticsearch用户组
				createGroupCommand := "sudo groupadd elasticsearch"
				output, err = utils.ExecCommand(createGroupCommand)
				if err != nil {
					message := fmt.Sprintf("创建elasticsearch用户组异常: %v, output: %s", err, output)
					logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
					return // 中断执行，避免后续步骤继续
				}
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "elasticsearch用户组创建成功", "Elasticsearch", "Server")
			}

			// 检查elasticsearch用户是否存在
			checkUserCommand := "id -u elasticsearch_logic"
			output, err = utils.ExecCommand(checkUserCommand)
			if err != nil || strings.TrimSpace(output) == "" {
				// 如果用户不存在，创建elasticsearch用户并加入elasticsearch组
				createUserCommand := "sudo useradd -g elasticsearch elasticsearch_logic"
				output, err = utils.ExecCommand(createUserCommand)
				if err != nil {
					message := fmt.Sprintf("创建elasticsearch用户异常: %v, output: %s", err, output)
					logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
					return // 中断执行，避免后续步骤继续
				}
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "elasticsearch用户创建成功", "Elasticsearch", "Server")
			}
			in.NodeInfo.Username = "elasticsearch_logic"
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查当前用户是否为root用户完成", "Elasticsearch", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组...", "Elasticsearch", "Server")
		chownCommand := fmt.Sprintf("sudo chown -R %s:%s %s", in.NodeInfo.Username, in.NodeInfo.Username, home)
		output, err = utils.ExecCommand(chownCommand)
		if err != nil {
			message := fmt.Sprintf("修改组件所属用户组异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组完成", "Elasticsearch", "Server")

		checkPathCommand = fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
sudo chown -R %s:%s %s
`, in.ElasticsearchConfig.PathData, in.ElasticsearchConfig.PathData, in.NodeInfo.Username, in.NodeInfo.Username, in.ElasticsearchConfig.PathData)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录...", "Elasticsearch", "Server")
		output, err = utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件数据存储目录异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录完成", "Elasticsearch", "Server")

		checkPathCommand = fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
sudo chown -R %s:%s %s
`, in.ElasticsearchConfig.PathLogs, in.ElasticsearchConfig.PathLogs, in.NodeInfo.Username, in.NodeInfo.Username, in.ElasticsearchConfig.PathLogs)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件日志存储目录...", "Elasticsearch", "Server")
		output, err = utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件日志存储目录异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件日志存储目录完成", "Elasticsearch", "Server")
		step++
		currentIP, currentIPErr := utils.GetCurrentInternalIP()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机IP: %v", in.NodeInfo.Host, currentIPErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		currentHostname, currentHostErr := utils.GetCurrentHostname()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机名: %v", in.NodeInfo.Host, currentHostErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}

		discoverySeedHosts, err := json.Marshal(in.ElasticsearchConfig.DiscoverySeedHosts)
		if err != nil {
			message := fmt.Sprintf("%s: DiscoverySeedHosts JSON Marshal error: %v", in.NodeInfo.Host, err)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}

		initialMasterNodes, err := json.Marshal(in.ElasticsearchConfig.InitialMasterNodes)
		if err != nil {
			message := fmt.Sprintf("%s: InitialMasterNodes JSON Marshal error: %v", in.NodeInfo.Host, err)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}

		nodeMasterFlag := true
		nodeDataFlag := true

		if value, exists := in.ElasticsearchConfig.NodeMaster[currentHostname]; exists {
			nodeMasterFlag = value
		} else if value, exists := in.ElasticsearchConfig.NodeMaster[currentIP]; exists {
			nodeMasterFlag = value
		}

		if value, exists := in.ElasticsearchConfig.NodeData[currentHostname]; exists {
			nodeDataFlag = value
		} else if value, exists := in.ElasticsearchConfig.NodeData[currentIP]; exists {
			nodeDataFlag = value
		}

		configFile := fmt.Sprintf(`
# ======================== Elasticsearch Configuration =========================
#
# NOTE: Elasticsearch comes with reasonable defaults for most settings.
#       Before you set out to tweak and tune the configuration, make sure you
#       understand what are you trying to accomplish and the consequences.
#
# The primary way of configuring a node is via this file. This template lists
# the most important settings you may want to configure for a production cluster.
#
# Please consult the documentation for further information on configuration options:
# https://www.elastic.co/guide/en/elasticsearch/reference/index.html
#
# Use a descriptive name for your cluster:
cluster.name: %s

node.name: %s

# Add custom attributes to the node:
#node.attr.rack: r1

# 设置索引数据的存储路径，默认是es根目录下的data文件夹，可以设置多个存储路径，用逗号隔开，例：/path/to/data1,/path/to/data2
path.data: %s

# 设置日志文件的存储路径，默认是es根目录下的logs文件夹
path.logs: %s

# 设置配置文件的存储路径，默认是es根目录下的config文件夹
path.conf: %s/config

bootstrap.memory_lock: %v

# 设置搜索线程池最大并发线程数，控制并发搜索请求处理能力
thread_pool.search.size: %d

# 设置为true来锁住内存不进行swapping。因为当jvm开始swapping时es的效率 会降低，所以要保证它不swap，可以把ES_MIN_MEM和ES_MAX_MEM两个环境变量设置成同一个值，并且保证机器有足够的内存分配给es。 同时也要允许elasticsearch的进程可以锁住内存
bootstrap.mlockall: %v

# Set the bind address to a specific IP (IPv4 or IPv6):
network.host: %s

http.port: %d

# 设置节点间通信使用的TCP端口，默认为9300                                               
transport.tcp.port: %d

# 设置是否压缩tcp传输时的数据，默认为false，不压缩
transport.tcp.compress: false

# 是否有资格竞选主节点
node.master: %v

# 该节点是否存储索引数据
node.data: %v

# 设置内容的最大容量，默认100mb
http.max_content_length: %s

# 是否使用http协议对外提供服务，默认为true，开启
http.enabled: %v

# gateway的类型，默认为local即为本地文件系统，可以设置为本地文件系统，分布式文件系统，hadoop的HDFS，和amazon的s3服务器等
gateway.type: %s

# 设置集群中N个节点启动时进行数据恢复，默认为1
gateway.recover_after_nodes: %d

# 设置初始化数据恢复进程的超时时间，默认是5分钟
gateway.recover_after_time: %s

# 设置这个集群中节点的数量，默认为2，一旦这N个节点启动，就会立即进行数据恢复
gateway.expected_nodes: %d

# 初始化数据恢复时，并发恢复线程的个数，默认为4
cluster.routing.allocation.node_initial_primaries_recoveries: %d

# 添加删除节点或负载均衡时并发恢复线程的个数，默认为4
cluster.routing.allocation.node_concurrent_recoveries: %d

# 设置数据恢复时限制的带宽，如入100mb，默认为0，即无限制
indices.recovery.max_size_per_sec: %s

# 设置这个参数来限制从其它分片恢复数据时最大同时打开并发流的个数，默认为5
indices.recovery.concurrent_streams: %d

# *表示支持所有域名
http.cors.allow-origin: "%s"

# 是否支持跨域
http.cors.enabled: %v

# 是否锁定Elasticsearch进程使用的内存，防止被操作系统交换到磁盘
bootstrap.system_call_filter: %v

# Pass an initial list of hosts to perform discovery when this node is started:
# The default list of hosts is ["127.0.0.1", "[::1]"]
# 列出初始发现节点列表，用于新节点加入集群时发现已有节点
discovery.seed_hosts: %s
#
# Bootstrap the cluster using an initial set of master-eligible nodes:
cluster.initial_master_nodes: %s

# Require explicit names when deleting indices:
action.destructive_requires_name: %v

# 启用X-Pack安全模块，提供身份验证、授权、加密等功能
xpack.security.enabled: %v

# 启用节点间通信的SSL加密，保护数据传输安全
xpack.security.transport.ssl.enabled: %v
`, in.ElasticsearchConfig.ClusterName,
			currentHostname,
			in.ElasticsearchConfig.PathData,
			in.ElasticsearchConfig.PathLogs,
			in.ElasticsearchConfig.PathConf,
			in.ElasticsearchConfig.BootstrapMemoryLock,
			in.ElasticsearchConfig.ThreadPoolSearchSize,
			in.ElasticsearchConfig.BootstrapMlockall,
			currentIP,
			in.ElasticsearchConfig.HttpPort,
			in.ElasticsearchConfig.TransportTcpPort,
			nodeMasterFlag,
			nodeDataFlag,
			in.ElasticsearchConfig.HttpMaxContentLength,
			in.ElasticsearchConfig.HttpEnabled,
			in.ElasticsearchConfig.GatewayType,
			in.ElasticsearchConfig.GatewayRecoverAfterNodes,
			in.ElasticsearchConfig.GatewayRecoverAfterTime,
			in.ElasticsearchConfig.GatewayExpectedNodes,
			in.ElasticsearchConfig.ClusterRoutingAllocationNodeInitialPrimariesRecoveries,
			in.ElasticsearchConfig.ClusterRoutingAllocationNodeConcurrentRecoveries,
			in.ElasticsearchConfig.IndicesRecoveryMaxSizePerSec,
			in.ElasticsearchConfig.IndicesRecoveryConcurrentStreams,
			in.ElasticsearchConfig.HttpCorsAllowOrigin,
			in.ElasticsearchConfig.HttpCorsEnabled,
			in.ElasticsearchConfig.BootstrapSystemCallFilter,
			discoverySeedHosts,
			initialMasterNodes,
			in.ElasticsearchConfig.ActionDestructiveRequiresName,
			in.ElasticsearchConfig.XpackSecurityEnabled,
			in.ElasticsearchConfig.XpackSecurityTransportSslEnabled)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 elasticsearch.yml 文件...", "Elasticsearch", "Server")
		configFileCommand := fmt.Sprintf("echo '%s' | sudo tee %s/config/elasticsearch.yml", configFile, home)
		output, err = utils.ExecCommand(configFileCommand)
		if err != nil {
			message := fmt.Sprintf("配置 elasticsearch_logic.yml 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置 elasticsearch.yml 文件完成", "Elasticsearch", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置 jvm.options 文件...", "Elasticsearch", "Server")
		configCommand := fmt.Sprintf(" sed -i '/^\\s*#\\?\\s*-Xms1g/c\\-Xms%s' %s/config/jvm.options", in.ElasticsearchConfig.JvmXms, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*-Xmx1g/c\\-Xmx%s' %s/config/jvm.options", in.ElasticsearchConfig.JvmXmx, home)
		output, err = utils.ExecCommand(configCommand)
		if err != nil {
			message := fmt.Sprintf("配置 jvm.options 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置 jvm.options 文件完成", "Elasticsearch", "Server")

		serviceFile := fmt.Sprintf(`[Unit]
Description=Elasticsearch
Documentation=https://www.elastic.co
Wants=network-online.target
After=network-online.target

[Service]
User=elasticsearch_logic
Group=elasticsearch_logic
Environment=JAVA_HOME=%s/jdk
Environment=ES_HOME=%s
Environment=ES_PATH_CONF=%s/config
Environment=PID_DIR=%s
Environment=ES_SD_NOTIFY=true
EnvironmentFile=-%s/config
WorkingDirectory=%s
ExecStart=%s/bin/elasticsearch -p ${PID_DIR}/elasticsearch.pid --quiet

# 指定此进程可以打开的最大文件数
LimitNOFILE=655355
# 指定此进程可以打开的最大进程数
LimitNPROC=655355
# 最大虚拟内存
LimitAS=infinity
# Specifies the maximum file size
LimitFSIZE=infinity
# 超时设置 0-永不超时
TimeoutStopSec=0
# SIGTERM是停止java进程的信号
KillSignal=SIGTERM
# Java process is never killed
SendSIGKILL=no
# When a JVM receives a SIGTERM signal it exits with code 143
SuccessExitStatus=143
# Allow a slow startup before the systemd notifier module kicks in to extend the timeout
TimeoutStartSec=75

[Install]
WantedBy=multi-user.target`, home, home, home, home, home, home, home)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 elasticsearch.service 文件...", "Elasticsearch", "Server")
		writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/elasticsearch.service", serviceFile)
		output, err = utils.ExecCommand(writeServiceFileCommand)
		if err != nil {
			message := fmt.Sprintf("创建 elasticsearch.service 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 elasticsearch.service 文件完成", "Elasticsearch", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Elasticsearch服务...", "Elasticsearch", "Server")
		reloadSystemdCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable elasticsearch.service --now")
		output, err = utils.ExecCommand(reloadSystemdCommand)
		if err != nil {
			message := fmt.Sprintf("启动Elasticsearch服务异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Elasticsearch服务完成", "Elasticsearch", "Server")
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待Elasticsearch服务完全启动", "Elasticsearch", "Server")

		time.Sleep(5 * time.Second)

		statusCommand := "sudo systemctl is-active elasticsearch.service"
		step++
		output, err = utils.ExecCommand(statusCommand)
		if err != nil || strings.TrimSpace(output) != "active" {
			message := fmt.Sprintf("Elasticsearch启动后未正常运行: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Elasticsearch", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "Elasticsearch部署成功", "Elasticsearch", "Server")

		elasticsearchInfo := &models.ComponentInfo{
			ClusterID:      in.ClusterID,
			ComponentName:  "Elasticsearch",
			Version:        in.Version,
			NodeHost:       currentHostname,
			NodeIP:         currentIP,
			ComponentRole:  "Server",
			HomePath:       home,
			DataStorageDir: in.ElasticsearchConfig.PathData,
			Port:           strconv.Itoa(int(in.ElasticsearchConfig.HttpPort)),
			Status:         true,
			AdditionalInfo: fmt.Sprintf("config_file=%s/config/elasticsearch.yml", home),
		}

		componentInfoManager.SendComponentInfo(elasticsearchInfo)

		//err = utils.LogComponentInfo(l.svcCtx.Conn, elasticsearchInfo)
		//if err != nil {
		//	logx.Errorf("%s 节点: 更新 %s %s 角色信息失败: %v", in.NodeInfo.Host, "Elasticsearch", "Server", err)
		//} else {
		//	logx.Infof("%s 节点: 更新 %s %s 角色信息成功", in.NodeInfo.Host, "Elasticsearch", "Server")
		//}

	}()
	wg.Wait()
}
